package com.yihaodian.search.broker;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.yihaodian.api.search.SearchRequest;
import com.yihaodian.api.search.SearchResponse;
import com.yihaodian.api.search.misc.SearchHit;
import com.yihaodian.api.search.misc.SearchSort;
import com.yihaodian.api.search.misc.SearchSortField;
import com.yihaodian.search.broker.shard.ShardSearchResponse;

public class SearchHitMerger {

	public static SearchHit[] merge(SearchRequest request,
			List<ShardSearchResponse> shardResponses) {
		java.util.PriorityQueue<SearchShardDoc> queue = new java.util.PriorityQueue<SearchShardDoc>();
		Map<String, SearchHit> searchHitMap = new HashMap<String, SearchHit>();

		// TODO, SCORE AS DEFAULT SORT
		int sortFieldSize = 0;
		SearchSort sort = request.getSort();
		SearchSortField[] sortFields = null;
		if (sort != null) {
			sortFields = sort.getFields();
			sortFieldSize = sortFields.length;
		}

		for (ShardSearchResponse rsp : shardResponses) {
			if (rsp.getSearchResponse() == null)
				continue;
			SearchHit[] hits = rsp.getSearchResponse().getHits();
			if (hits == SearchResponse.NO_HITS)
				continue;
			String shardId = rsp.getShardId();

			for (int i = 0; i < hits.length; i++) {
				SearchHit hit = hits[i];
				searchHitMap.put(shardId + "_" + hit.getDocId(), hit);

				Object[] values = hit.getSortValues();
				SortValue[] sortValues = new SortValue[sortFieldSize + 1];
				if (sortFields != null) {
					for (int j = 0; j < sortFieldSize; j++) {
						sortValues[j] = new SortValue((Comparable) values[j],
								sortFields[j].isReverse());
					}
				}
				/**
				 * score 必须考虑 docId 可以不考虑
				 */
				sortValues[sortFieldSize] = new SortValue(hit.getScore(), true);
				SearchShardDoc doc = new SearchShardDoc(hit.getDocId(),
						shardId, i, sortValues);
				queue.add(doc);
			}
		}
		int resultSize = queue.size() - request.getOffset();
		if (resultSize <= 0)
			return SearchResponse.NO_HITS;
		resultSize = Math.min(request.getCount(), resultSize);
		SearchHit[] returnHit = new SearchHit[resultSize];
		int offset = request.getOffset();
		for (int i = 0; i < resultSize + request.getOffset(); i++) {
			if (i < request.getOffset()) {
				queue.poll();
				continue;
			}
			SearchShardDoc shardDoc = queue.poll();
			SearchHit hit = searchHitMap.get(shardDoc.shardId + "_"
					+ shardDoc.docId);
			if (hit != null) {
				returnHit[i-offset] = hit;
			}
		}
		return returnHit;
	}

	static class SortValue {
		Comparable value;
		boolean reverse;

		public SortValue(Comparable value, boolean reverse) {
			this.value = value;
			this.reverse = reverse;
		}
	}

	static class SearchShardDoc implements Comparable<SearchShardDoc> {
		int docId;
		String shardId;
		int orderInShard;
		SortValue[] sortValues;

		public SearchShardDoc(int docId, String shardId, int orderInShard,
				SortValue[] sortValues) {
			this.docId = docId;
			this.shardId = shardId;
			this.orderInShard = orderInShard;
			this.sortValues = sortValues;
		}

		@Override
		public int compareTo(SearchShardDoc o) {
			if (this.shardId == o.shardId) {
				return this.orderInShard - o.orderInShard;
			}
			int c = 0;
			if (this.sortValues != null && o.sortValues != null) {
				for (int i = 0; i < this.sortValues.length; i++) {
					SortValue aSortValue = this.sortValues[i];
					SortValue bSortValue = o.sortValues[i];
					c = (aSortValue.reverse) ? bSortValue.value
							.compareTo(aSortValue.value) : aSortValue.value
							.compareTo(bSortValue.value);
					if (c != 0) {
						return c;
					}
				}
			}
			return c;
		}

	}

}
